package com.lxn.until

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import com.lxn.until.MyConfig.{ES_HOST, ES_PORT}
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.indices.GetIndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.search.builder.SearchSourceBuilder

import scala.collection.mutable.ListBuffer

object MyEsUtils {

  //向es中写入数据 批量写入

  //创建连接对象
  val esClient: RestHighLevelClient = build()

  def build(): RestHighLevelClient = {
    val host: String = MyPropertyUtils(MyConfig.ES_HOST)
    val port: String = MyPropertyUtils(MyConfig.ES_PORT)
    val httpHost: HttpHost = new HttpHost(host, port.toInt)

    val builder: RestClientBuilder = RestClient.builder(httpHost)
    val client = new RestHighLevelClient(builder)
    client
  }

  def close(): Unit = {
    if(esClient!=null) esClient.close()
  }

  //批量插入
  //根据传入的索引和数据，插入,由于我们之前完成的操作只是至少一次消费，我们在这存储数据的时候通过幂等操作实现精确一次消费
  //索引我们需要传入id 和请求体数据
  def bulkSave(indexName : String,docs : List[(String , AnyRef)] ): Unit = {
    //创建批量请求对象
    val bulkRequest: BulkRequest = new BulkRequest(indexName)
    //开始进行批量插入操作
    for((docId,docObj) <- docs){
      //对每一个插入请求进行处理
      val indexRequest: IndexRequest = new IndexRequest()
      //指定插入的id
      indexRequest.id(docId)
      //将数据封装至source
      //首先转换数据格式
      val indexJsonObj: String = JSON.toJSONString(docObj, new SerializeConfig(true))
      indexRequest.source(indexJsonObj,XContentType.JSON)
      //将每一个插入请求放入批量处理里面
      bulkRequest.add(indexRequest)
    }
    //执行批量请求
    esClient.bulk(bulkRequest,RequestOptions.DEFAULT)
  }


  //根据字段查询值
  def searchField(indexName: String, fieldName: String): List[String] = {
    //一般而言需要判断索引是否存在，再进行查询
    val getIndexRequest = new GetIndexRequest(indexName)
    //判断索引是否存在
    val isExists: Boolean = {
      esClient.indices().exists(getIndexRequest,RequestOptions.DEFAULT)
    }
    if(!isExists){
      return null
    }

    //查询字段
    //创建存储结果的集合  这是一个可变集合
    val mids: ListBuffer[String] = ListBuffer[String]()
    //创建一个search请求
    val searchRequest: SearchRequest = new SearchRequest(indexName)

    val searchSourceBuilder: SearchSourceBuilder = new SearchSourceBuilder()
    //封装请求  es默认只能获取10条数据，我们可以在这里修改
    searchSourceBuilder.fetchSource(fieldName, null).size(10000)
    //将查询请求放入source里面
    searchRequest.source(searchSourceBuilder)
    val searchResponse: SearchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT)

    //获取执行结果
    val hits = searchResponse.getHits.getHits
    for (hit <- hits) {
      //获取结果
      val sourceAsMap = hit.getSourceAsMap
      //根据key获取值
      val mid = sourceAsMap.get(fieldName).toString
      //存储值
      mids.append(mid)
    }
    //返回不可变集合
    mids.toList
  }

  def main(args: Array[String]): Unit = {
    println(searchField("gmall_dau_info2023-07-11","mid"))
  }

}
